package RabbitMQ

import (
	"fmt"
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

const MQURL = "amqp://imoocuser:imoocuser@127.0.0.1:5672/imooc"

type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	// 队列名称
	QuequName string
	// 交换机
	Exchange string
	// KEY
	Key string
	// 连接信息
	Mqurl string
	// 交换机类型
	ExchangeType string
}

// 创建RabbitM结构体实例
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	var err error
	// 创建连接

	rabbitmq := &RabbitMQ{
		QuequName: queueName,
		Exchange:  exchange,
		Key:       key,
		Mqurl:     MQURL,
	}
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	rabbitmq.failOnErr(err, "创建连接错误")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "获取channel错误")
	return rabbitmq
}

// 断开channel 和 connection
func (r *RabbitMQ) Destory() {
	r.channel.Close()
}

// 错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

// 创建简单模式下RabbitMQ实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
	return NewRabbitMQ(queueName, "", "")
}

// 简单模式 Step: 1. 创建简单模式下的RabbitMQ实例
func (r *RabbitMQ) PublishSimple(message string) {
	// 申请队列，如果队列不存在会自动创建，如果存在则跳过创建
	// 保证队列存在，消息能发送到队列中
	_, err := r.channel.QueueDeclare(
		r.QuequName,
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		false,
		// 是否阻塞
		false,
		// 额外苏醒
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	// 发送消息到队列中
	r.channel.Publish(
		r.Exchange,
		r.QuequName,
		// 如果为true,根据exchange 类型和routekey规则，如果无法找到符合条件的队列，那么会吧发送的消息返回给发送者
		false,
		// 如果为true,当exchange发消息到队列后发现队列上没有绑定消费者，则会把消息发给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// 简单模式 Step: 2. 简单模式下的生成代码
func (r *RabbitMQ) ConsumeSimple() {
	// 申请队列，如果队列不存在会自动创建，如果存在则跳过创建
	// 保证队列存在，消息能发送到队列中
	_, err := r.channel.QueueDeclare(
		r.QuequName,
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		false,
		// 是否阻塞
		false,
		// 额外苏醒
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	msgs, err := r.channel.Consume(
		r.QuequName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
			// 手动应答
			if err := d.Ack(false); err != nil {
				log.Fatalf("Failed to acknowledge a message: %v", err)
			}
		}
	}()
	log.Printf("[*] Waiting for messages,To exit press CTRL+C")
	<-forever
}

// 订阅模式创建RabbitMQ实例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, "")
	rabbitmq.ExchangeType = "fanout"
	return rabbitmq
}

// 订阅者模式生产
func (r *RabbitMQ) PublishPub(message string) {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	err = r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	r.failOnErr(err, "Failed to declare an exchange")

}

// 订阅者模式消费端

func (r *RabbitMQ) ReceiveSub() {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	// 试探性创建队列
	q, err := r.channel.QueueDeclare(
		"", // 随机生成队列名称
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		true, // 独占队列
		// 是否阻塞
		false,
		// 额外苏醒
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	// 绑定队列到exchange 中
	err = r.channel.QueueBind(
		q.Name,
		// 在pub/sub模式下 这里的key要为空
		"",
		r.Exchange,
		false,
		nil)
	r.failOnErr(err, "Failed to bind a queue")
	msgs, err := r.channel.Consume(
		q.Name,
		"",
		true, // 设置为 false 以启用手动应答
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
			// 手动应答
			// if err := d.Ack(false); err != nil {
			// 	log.Fatalf("Failed to acknowledge a message: %v", err)
			// }
		}
	}()
	log.Printf("[*] Waiting for messages,To exit press CTRL+C")
	<-forever

}

// 创建路由模式下RabbitMQ实例
func NewRabbitMQRouting(exchangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, routingKey)
	rabbitmq.ExchangeType = "direct"
	return rabbitmq
}

// 路由模式发送者
func (r *RabbitMQ) PublishRouting(message string) {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	err = r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	r.failOnErr(err, "Failed to Publishing")

}

// 订阅者模式消费端
// 在 RabbitMQ 包中

func (r *RabbitMQ) ReceiveRouting() {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	// 试探性创建队列
	q, err := r.channel.QueueDeclare(
		"", // 随机生成队列名称
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		true, // 独占队列
		// 是否阻塞
		false,
		// 额外苏醒
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	fmt.Println(r.Key)
	// 绑定队列到exchange 中
	err = r.channel.QueueBind(
		q.Name,
		// 在pub/sub模式下 这里的key要为空
		r.Key,
		r.Exchange,
		false,
		nil)
	r.failOnErr(err, "Failed to bind a queue")
	msgs, err := r.channel.Consume(
		q.Name,
		"",
		true, // 设置为 false 以启用手动应答
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
			// 手动应答
			// if err := d.Ack(false); err != nil {
			// 	log.Fatalf("Failed to acknowledge a message: %v", err)
			// }
		}
	}()
	log.Printf("[*] Waiting for messages,To exit press CTRL+C")
	<-forever

}

// 创建Topic模式下RabbitMQ实例
func NewRabbitMQTopic(exchangeName string, routingKey string) *RabbitMQ {
	rabbitmq := NewRabbitMQ("", exchangeName, routingKey)
	rabbitmq.ExchangeType = "topic"
	return rabbitmq
}

// 话题模式发送消息
func (r *RabbitMQ) PublishTopic(message string) {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	err = r.channel.Publish(
		r.Exchange,
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	r.failOnErr(err, "Failed to Publishing")
}

// 话题模式接受消息
// 要注意key,规则，其中 * 用于匹配一个单词，# 用于匹配多个单词 可以是0个
// 匹配 imooc.* 表示匹配 imooc.hello 但是 imooc.hello.one 需要用 imooc.# 才能匹配到
func (r *RabbitMQ) ReceiveTopic() {
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		r.ExchangeType,
		true,
		false,
		//  true 表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")
	// 试探性创建队列
	q, err := r.channel.QueueDeclare(
		"", // 随机生成队列名称
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		true, // 独占队列
		// 是否阻塞
		false,
		// 额外苏醒
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")
	fmt.Println(r.Key)
	// 绑定队列到exchange 中
	err = r.channel.QueueBind(
		q.Name,
		// 在pub/sub模式下 这里的key要为空
		r.Key,
		r.Exchange,
		false,
		nil)
	r.failOnErr(err, "Failed to bind a queue")
	msgs, err := r.channel.Consume(
		q.Name,
		"",
		true, // 设置为 false 以启用手动应答
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	go func() {
		for d := range msgs {
			log.Printf("Received a message: %s", d.Body)
			// 手动应答
			// if err := d.Ack(false); err != nil {
			// 	log.Fatalf("Failed to acknowledge a message: %v", err)
			// }
		}
	}()
	log.Printf("[*] Waiting for messages,To exit press CTRL+C")
	<-forever
}
